package net.wicp.tams.common.binlog.self.event.rows;

import java.util.ArrayList;
import java.util.List;

import org.apache.commons.collections.CollectionUtils;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.Conf;
import net.wicp.tams.common.Result;
import net.wicp.tams.common.apiext.ByteUtil;
import net.wicp.tams.common.apiext.StringUtil;
import net.wicp.tams.common.binlog.self.bean.EventBean;
import net.wicp.tams.common.binlog.self.bean.Host;
import net.wicp.tams.common.binlog.self.bean.SendBean;
import net.wicp.tams.common.binlog.self.bean.SendRow;
import net.wicp.tams.common.binlog.self.bean.TableMapBean;
import net.wicp.tams.common.binlog.self.callback.ICol;
import net.wicp.tams.common.binlog.self.event.AbsEvent;
import net.wicp.tams.common.binlog.self.sender.ISender;
import net.wicp.tams.common.constant.OptType;
import net.wicp.tams.common.thread.threadlocal.PerthreadManager;

@SuppressWarnings("rawtypes")
@Slf4j
public abstract class RowsEvent extends AbsEvent {
	private static int maxRows = Integer.parseInt(StringUtil.hasNull(Conf.get("zorro.sender.max"), "400"));
	protected int version = 1;// 只支持版本为1/2的事件,默认为1
	protected long tableId;
	protected int flags;
	// protected String extraData;
	protected long columnCount;
	protected byte[] columnsPresentBitmap1;
	protected byte[] columnsPresentBitmap2;
	protected TableMapBean table;
	protected OptType optType;

	public RowsEvent(EventBean event) {
		super(event);
	}

	@Override
	public final Result parseBody(ISender... senders) {
		// Long time = System.currentTimeMillis();

		// rowlog("RowsEvent的body位置");
		int postHeadLength = event.getHead().getEventType().getPostHeaderLength();
		if (postHeadLength == -1) {
			return Result.getError("没有初始化，FormatDescription事件必须第一个读.");
		}
		if (postHeadLength == 6) {
			tableId = super.logBuffer.getInt32();// ByteUtil.readLongL(assitRead.readBytes(4));
		} else {
			tableId = super.logBuffer.getLong48();// ByteUtil.readLongL(assitRead.readBytes(6));
		}
		// 过滤表
		Host hostpre = (Host) PerthreadManager.getInstance().createValue("zorro-host").get();
		table = hostpre.findTableMapBean(tableId);
		if (table == null) {
			return Result.getError("no table" + tableId);// 没有table信息,在解析row信息前需要解析
			// TABLE_MAP_EVENT事件
		}

		// end 过滤表
		flags = super.logBuffer.getInt16();// ByteUtil.readIntL(assitRead.readBytes(2));
		if (version == 2) {
			int extraDataLength = super.logBuffer.getInt16();// ByteUtil.readIntL(assitRead.readBytes(2));
			super.logBuffer.forward(extraDataLength - 2);
			// extraData =
			// ByteUtil.readString(assitRead.readBytes(extraDataLength - 2));
		}

		// body
		columnCount = super.logBuffer.getPackedLong();// assitRead.readUnsignedLong();
		int presentBitmap1Length = ((int) columnCount + 7) / 8;

		columnsPresentBitmap1 = new byte[presentBitmap1Length];
		this.logBuffer.fillBytes(columnsPresentBitmap1, 0, presentBitmap1Length);
		if (optType == OptType.update) {
			columnsPresentBitmap2 = new byte[presentBitmap1Length];
			super.logBuffer.fillBytes(columnsPresentBitmap2, 0, presentBitmap1Length);
			// columnsPresentBitmap2 =
			// assitRead.readBytes(presentBitmap1Length);
		}

		// rows:
		// rowlog("RowsEvent的rows位置");
		List<SendRow> rows = new ArrayList<>();
		// LogBackUtil.logDebugTime(log, "更新前置", time);
		while (super.logBuffer.hasRemaining()) {
			// time = System.currentTimeMillis();
			// Long time2 = System.currentTimeMillis();
			List<ICol> row1 = paseRow(columnsPresentBitmap1, true);
			// LogBackUtil.logDebugTime(log, "更新1", time);
			List<ICol> row2 = null;
			if (optType == OptType.update) {
				// time = System.currentTimeMillis();
				row2 = paseRow(columnsPresentBitmap2, false);
				// LogBackUtil.logDebugTime(log, "更新2", time);
			}
			// time = System.currentTimeMillis();
			SendRow sendRow = SendRow.builder().before(row1).after(row2).build();
			rows.add(sendRow);
			if (rows.size() >= maxRows) {
				// time = System.currentTimeMillis();
				doSend(rows, senders);
				// LogBackUtil.logDebugTime(log, "发送", time);
				rows = new ArrayList<>();
			}
			// LogBackUtil.logDebugTime(log, "发送", time);
			// LogBackUtil.logDebugTime(log, "整个", time2);
		}
		if (CollectionUtils.isNotEmpty(rows)) {
			doSend(rows, senders);
		}

		return Result.getSuc();
	}

	private List<ICol> paseRow(byte[] columnsPresentBitmap, boolean isBefore) {
		int presentBitmapNum = ByteUtil.byteshas1(columnsPresentBitmap);
		// presentBitmapNum = presentBitmapNum > columnCount.intValue() ?
		// columnCount.intValue() : presentBitmapNum;
		int len = (presentBitmapNum + 7) / 8;
		byte[] nulbitmap = super.logBuffer.getData(len);
		// byte[] nulbitmap = assitRead.readBytes((presentBitmapNum + 7) / 8);
		// 1、判断columnsPresentBitmap1 为1
		// int PresentIs = -1;
		List<ICol> row1 = new ArrayList<>();
		for (int i = 0; i < table.getColumnCount(); i++) {
			if ((columnsPresentBitmap[i / 8] & 0x01 << i % 8) != 0x00) {
				// PresentIs++;
				boolean isnull = (nulbitmap[i / 8] & 0x01 << (i % 8)) != 0x00;// 0
																				// 表示不是空值
				ICol col = paseCol(table.getColumnTypes()[i], table.getCols().get(i), table.getMetaDefs()[i], isnull);
				row1.add(col);
				if (col == null) {
					log.error("第[{}]列col为空值，binlog is [{}]", i, isnull);
				} else {
					if (i == 0 && isBefore) {// 打印第1列值，一般第一列为主键
						log.info("col0:[{}],value:[{}]", table.getCols().get(i), col.getClient());
					}
				}
			} else {// 没有持有
			}
		}
		return row1;
	}

	private Result doSend(List<SendRow> rows, ISender... senders) {
		Result ret = Result.getSuc();
		for (ISender sender : senders) {
			try {
				SendBean sendBean = SendBean.builder().columnCount((int) columnCount).rows(rows).gtid(event.getGtids())
						.tableMapBean(table).optType(optType).timestamp(event.getHead().getTimestamp()).build();
				Result rettemp = sender.sendMsg(sendBean);
				if (!rettemp.isSuc()) {
					ret = rettemp;
				}
			} catch (Throwable e) {
				ret = Result.getError("失败:" + e.getMessage());
				log.error("发送者调用失败", e);
			}
		}
		if (senders.length == 1 && !ret.isSuc()) {
			while (true) {
				try {
					log.error("唯一的发送者失败,暂停进程3秒，事件位置[{}]", row(event.getBeginHead()));
					Thread.sleep(3 * 1000);// 唯一的发送者失败,暂停进程，3
					ret = doSend(rows, senders);
					if (ret.isSuc()) {
						break;
					}
				} catch (InterruptedException e) {
					log.error("唯一的发送者失败，休息3秒时出错");
				}
			}
		}
		return ret;
	}

	public OptType getOptType() {
		return optType;
	}

}
